package com.you.rpc;

import com.you.rpc.base.ObjectDecoder;
import com.you.rpc.base.ObjectEncoder;
import com.you.rpc.client.DefaultInvokeFuture;
import com.you.rpc.client.InvokeFuture;
import com.you.rpc.entity.RequestCommand;
import com.you.rpc.entity.ResponseCommand;
import com.you.rpc.entity.ResponseStatus;
import com.you.rpc.entity.RpcAddress;
import com.you.rpc.exc.RpcRemotingException;
import com.you.rpc.client.ClientHandler;
import com.you.rpc.until.AddressUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class RpcClient {
    /**
     * 工作线程组:CPU可用核数*2
     */
    private NioEventLoopGroup group = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
    private Bootstrap bootstrap = new Bootstrap();
    /**
     * 请求地址(主机名:端口号),对应的ChannelFuture
     */
    private Map<RpcAddress, ChannelFuture> connections = new ConcurrentHashMap<>();
    /**
     * 请求ID,对应的任务序列
     */
    private Map<Integer, InvokeFuture> invokeFutureMap = new ConcurrentHashMap<>();

    /**
     * 多线程处理请求
     */
//    private static ExecutorService executorService = Executors.newFixedThreadPool(8);
    public void start() {
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new ObjectEncoder());
                        pipeline.addLast(new ObjectDecoder());
                        pipeline.addLast(new ClientHandler(connections, invokeFutureMap));
                    }
                });
        log.info("RPC Client启动...");
    }


    public void connect(String address) throws InterruptedException {
        connect(AddressUtil.turnToAddress(address));
    }

    private void connect(RpcAddress address) throws InterruptedException {
        if (connections.containsKey(address)) {
            return;
        }
        CountDownLatch latch = new CountDownLatch(1);
        ChannelFuture channelFuture = bootstrap.connect(address.getHost(), address.getPort());
        channelFuture.addListener((c) -> {
            if (c.isSuccess()) {
                connections.put(address, channelFuture);
                log.info("RPC Client连接{}成功!", address);
                latch.countDown();
            }
        });
        latch.await(3000, TimeUnit.MILLISECONDS);
    }


    public Object invokeSync(String address, Object request) throws Exception {
        return invokeSync(address, request, 3);
    }


    /**
     * 如果服务器地址未连接,连接服务器后再后再调用请求
     */
    public Object invokeSync(String address, Object request, int timeout) throws Exception {
        RpcAddress rpcAddress = AddressUtil.turnToAddress(address);
        if (connections.get(rpcAddress) == null) {
            synchronized (this) {
                if (connections.get(rpcAddress) == null) {
//                    log.info("<==@尝试创建新的连接@==>");
                    connect(rpcAddress);
                }
            }
        }
        return invokeSyncSimply(rpcAddress, request, timeout);
    }


    private Object invokeSyncSimply(RpcAddress address, Object request, int timeout) throws Exception {
        ChannelFuture channelFuture = connections.get(address);
        if (channelFuture == null) {
            throw new RpcRemotingException();
        }
        //随机生成请求id
        final int id = ThreadLocalRandom.current().nextInt();
        InvokeFuture invokeFuture = new DefaultInvokeFuture();
        invokeFutureMap.putIfAbsent(id, invokeFuture);
        //发送请求
        RequestCommand requestCommand = new RequestCommand();
        requestCommand.setRequestBody(request);
        requestCommand.setId(id);
        channelFuture.channel().writeAndFlush(requestCommand);
        //等待响应
        ResponseCommand responseCommand = invokeFuture.waitResponse(timeout, TimeUnit.SECONDS);
        if (responseCommand.getStatus() == ResponseStatus.SUCCESS) {
            return responseCommand.getResponseBody();
        }
        return null;
    }


    public void destroy() throws InterruptedException {
        for (Map.Entry<RpcAddress, ChannelFuture> entry : connections.entrySet()) {
            entry.getValue().channel().closeFuture().sync();
        }
        group.shutdownGracefully();
        log.info("RPC Client关闭");
    }
}
